package com.donglink.test.logmonitor.bolt;

import com.donglink.test.logmonitor.dao.MybatisDriver;
import com.donglink.test.logmonitor.entry.App;
import com.donglink.test.logmonitor.entry.Rule;
import com.donglink.test.logmonitor.service.FilterService;
import com.donglink.test.logmonitor.service.impl.FilterServiceImpl;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.IWindowedBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

public class FilterBolt extends BaseRichBolt {
    FilterService filterService;
    OutputCollector outputCollector;
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

        filterService=new FilterServiceImpl();
        this.outputCollector=outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {

        String log=tuple.getString(0);
        App app=filterService.isUnqualified(log);
        if(app!=null)
        {
            //如果收到监控
            outputCollector.emit(new Values(log,app));
        }
        else
        {
            //合格直接丢掉
            return;
        }


            MybatisDriver.pullAllData();


    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("log","app"));
    }
}
